#include	"a_global_vars.h"
#include	"config_parser.h"

void look4highest(int start, int end) {
	int i;
	//	printf("hd+1: %d	 gapEnd+1: %d\n", start, end);
	for (i = start; i < end; i++) {
		if (data[(i % INTERNAL_BUFFER_SIZE)].deliver)
			a_interval_performance->highest_deliverable_inst_id = i;
		else
			break;
	}
}

void execute_phase_2_b(paxos_msg *msg) {

	int i = msg->instance_number;
	int r = msg->round_number;

	if (data[(i % INTERNAL_BUFFER_SIZE)].packet.pack_id == data[(i % INTERNAL_BUFFER_SIZE)].instance_id + 1 && data[(i % INTERNAL_BUFFER_SIZE)].received) {
//		if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].rnd <= r) {
	//		if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd != r) {
//				paxos_instances[(i % INTERNAL_BUFFER_SIZE)].rnd = r;
//				paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd = r;
				msg->acceptor_count++;
				msg->type = PHASE_2_B;

//			} else {
//				if (paxos_instances[(i % INTERNAL_BUFFER_SIZE)].vrnd == r) { // This is a hack: we assume the value proposed is the same...

//					msg->acceptor_count++;
//					msg->type = PHASE_2_B;
//				}
//			}
		//}
	}
}

void acceptor_msg_type_recognizer(int sd, short event, void * arg) {

	int status;
	inst_paxos inst_paxos_messg;
	status = recv(sd, &inst_paxos_messg, sizeof(inst_paxos), MSG_WAITALL);
	assert(status == sizeof(inst_paxos));
	a_interval_performance->interval_bytes_received += status;
	a_handel_msg_inst_pack(&inst_paxos_messg);
}

void a_handel_msg_paxos(paxos_msg *m) {

	a_interval_performance->interval_num_paxos_packets_received++;

	if (m->type == PHASE_2_B) {
		execute_phase_2_b(m);

		if (PROCESS_ID == get_config_info()->total_num_acceptors - 1) {// lost acceptor, is the one who finds out the hD
			if (m->acceptor_count == N) {
				if (data[(m->instance_number % INTERNAL_BUFFER_SIZE)].received) {
					acceptor_0_deliver(m->instance_number, m->VALUE);// what to happen here?
					if (m->instance_number > gapEnd)
						gapEnd = m->instance_number;
					look4highest(a_interval_performance->highest_deliverable_inst_id + 1, gapEnd + 1);
					learner_deliver();
				}
			}
		}
	}

}

void a_handel_msg_inst_pack(inst_paxos * inst_paxos_messg) {

	int status = -1;
	int self_id = get_config_info()->self_acceptor_id;

	inst_paxos_messg->inst_packet.packet.p_msgs[0].num_nodes_visited++;

//	printf("%d\n", inst_paxos_messg->inst_packet.packet.p_msgs[0].msg_id);
	handle_packet(&(inst_paxos_messg->inst_packet)); //handles and updates the packet
	a_handel_msg_paxos(&(inst_paxos_messg->paxos_2b));

	if (PROCESS_ID == (get_config_info()->total_num_acceptors - 1)) {
		if (inst_paxos_messg->inst_packet.packet.p_msgs[0].num_nodes_visited == get_config_info()->ring_size) {
			short_dec_notification short_decision;
			short_decision.highestDeliverable = a_interval_performance->highest_deliverable_inst_id;

			if (inst_paxos_messg->paxos_2b.acceptor_count == N) {
				short_decision.instance_id = inst_paxos_messg->inst_packet.instance_id;
				short_decision.location_index_in_proposer = inst_paxos_messg->inst_packet.packet.p_msgs[0].location_index_in_proposer;
				short_decision.pack_id = inst_paxos_messg->inst_packet.packet.pack_id;
				short_decision.proposer_id = inst_paxos_messg->inst_packet.packet.p_msgs[0].proposer_id;
				short_decision.proposer_msg_id = inst_paxos_messg->inst_packet.packet.p_msgs[0].msg_id;
			} else
				short_decision.instance_id = -1;
			short_decision.type = 's';
			status = sendto((get_config_info()->server_infos[self_id].successor_sid), &(short_decision), sizeof(short_dec_notification), 0, (struct sockaddr*) &(get_config_info()->server_infos[self_id].ring_successor_addr), sizeof(struct sockaddr_in));
			assert(status == sizeof(short_dec_notification));
			a_interval_performance->interval_bytes_sent += status;

		} else {
			inst_paxos_messg->inst_packet.highestDeliverable = a_interval_performance->highest_deliverable_inst_id;
			inst_paxos_messg->inst_packet.type = 'i';
//			printf("msg_id: %d	nodes: %d ring: %d\n", inst_paxos_messg->inst_packet.packet.p_msgs[0].msg_id, inst_paxos_messg->inst_packet.packet.p_msgs[0].num_nodes_visited, get_config_info()->ring_size);
			status = send((get_config_info()->server_infos[self_id].successor_sid), &(inst_paxos_messg->inst_packet), sizeof(inst), MSG_WAITALL);
			assert(status == sizeof(inst));

			a_interval_performance->interval_bytes_sent += status;
		}
	} else {
		inst_paxos_messg->type = 'n';
		status = send((get_config_info()->server_infos[self_id].successor_sid), inst_paxos_messg, sizeof(inst_paxos), MSG_WAITALL);
		assert(status == sizeof(inst_paxos));
		a_interval_performance->interval_bytes_sent += status;
		//learner_deliver();
	}

}

void handle_packet(inst *new_data) {

	rtag = true;

	if (a_interval_performance->highest_deliverable_inst_id < new_data->highestDeliverable) {
		a_interval_performance->highest_deliverable_inst_id = new_data->highestDeliverable;
		learner_deliver();
	}
	memcpy(&data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)], new_data, sizeof(inst));
	data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].received = true;
	data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].deliver = false;
	data[(new_data->instance_id % INTERNAL_BUFFER_SIZE)].highestDeliverable = -1;
	a_interval_performance->interval_num_packets_received++;

}
